package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.table.expressions.In;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.concurrent.TimeUnit;

public class MyWindowTest {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT,8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.setParallelism(1);

        SingleOutputStreamOperator<Integer> dataStreamSource = env.addSource(new UdfSourceTest()).returns(Integer.class);
        SingleOutputStreamOperator<Tuple2<Integer, Integer>> mapStream = dataStreamSource.map(x -> Tuple2.of(1, x))
                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                }));

        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = mapStream.keyBy(t -> t.f0);

        //滚动窗口
        SingleOutputStreamOperator tumblingSum = keyedStream
                .window(TumblingProcessingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
                .sum(1);
        tumblingSum.print("TumblingSum:");




        //滑动窗口
        SingleOutputStreamOperator slidingSum = keyedStream
                .window(SlidingProcessingTimeWindows.of(Time.of(5, TimeUnit.SECONDS),Time.of(1, TimeUnit.SECONDS)))
                .sum(1);
        slidingSum.print("SlidingSum:");
        //会话窗口
        SingleOutputStreamOperator sessionSum = keyedStream
                .window(ProcessingTimeSessionWindows.withGap(Time.of(5, TimeUnit.SECONDS)))
                .sum(1);
        sessionSum.print("SessionSum:");
        //计数滚动窗口
        keyedStream.countWindow(3).sum(1).print("tumblingCountWindow");
        //计数滑动窗口
        keyedStream.countWindow(3,1).sum(1).print("slidingCountWindow");

        keyedStream.countWindow(3,1).process(new ProcessWindowFunction<Tuple2<Integer, Integer>, String, Integer, GlobalWindow>() {
            @Override
            public void process(Integer integer, ProcessWindowFunction<Tuple2<Integer, Integer>, String, Integer, GlobalWindow>.Context context, Iterable<Tuple2<Integer, Integer>> elements, Collector<String> out) throws Exception {
               String s ="elements:";
                Iterator<Tuple2<Integer, Integer>> iterator = elements.iterator();
                while (iterator.hasNext()){
                    s=s+iterator.next().f1;
                }
                out.collect(s);
            }
        }).print("slidingCountWindow");

        //allWindow
        mapStream.countWindowAll(3, 1).sum(0).print("allWindow");


        env.execute();
    }
}
